package com.yanxu;

import com.yanxu.domain.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 */
public class Api02_Source {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子

        //1：从文件中读取源数据
        DataStreamSource<String> streamSource1 = env.readTextFile("input/clicks.csv");
        streamSource1.print("from file >>>");

        //2：从socket中读取源数据
        DataStreamSource<String> streamSource2 = env.socketTextStream("192.168.1.30", 6666);
        streamSource2.print("from socket >>>");

        //3：从集合中创建数据流
        ArrayList<Event> events = new ArrayList<>();
        events.add(new Event("Mary", "./home", 1000L));
        events.add(new Event("Bob", "./cart", 2000L));
        DataStreamSource<Event> streamSource3 = env.fromCollection(events);
        streamSource3.print("from collection >>>");

        //4: 从给定 elements 元素创建数据流
        DataStreamSource<Event> streamSource4 = env.fromElements(
                new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L)
        );
        streamSource4.print("from Elements >>>");

        /*
        * 注意：
        *  fromCollection：接受一个集合创建 DataStream, 适合处理内存中的数据，只能是 单并行算子
        *  fromElements：从一系列元素创建 DataStream, 更灵活，数据同样保存在内存中，但是不显示并行度。 适用于少量静态的数据。
        * */


        //对环境进行执行。
        env.execute();
    }
}